package com.roy.kafka.trident;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.Func;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff;
import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff.TimeInterval;
import org.apache.storm.kafka.spout.KafkaSpoutRetryService;
import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutOpaque;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.io.Serializable;
import java.util.List;
import java.util.concurrent.TimeUnit;

import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST;

/**
 * 据网上介绍，这种Trident方式构建拓扑比Storm原有方式构建拓扑在事务安全性等方面有更高的保障。
 * 使用中发现，这种应该还是类似于window式的窗口操作。可以将结果分区划分成一个一个的小Batch
 * 然后提供了对消息进行小Batch局部整合再全量整合的处理方式。
 *
 * 这个示例的测试方式是这样的：先手动往kafka的track topic中发送消息，例如 "a a a a b b b b c c c c c"
 * 然后启动应用。这些kafka的历史消息就会被消费到TridentState中。
 * 在最后的localSubmitter.printResults 方法中就会启动几次DRPC调用，统计每个字符出现的次数
 *
 * 注意，DRPC服务是Storm的分布式RPC调用服务。 tridentTopology.newDRPCStream 这个是启动一个 RPC 服务端，只要拓扑在执行，就会一直启动。
 * 而DRPC的客户端调用是另外手动调用的。这个示例中是在LocalSubmitter的print中调用的。drpc.execute("words", "a b c d e f")
 *
 * 另外，tridentTopology.newStream 这个是会启动一个流，这个流只要拓扑在启动，就会不停的发射tuple的。要跟DRPCStream区分开理解。
 */
public class TridentKafkaTopology {

    public static final String TOPIC1="track";
    public static final String TOPIC2="track2";
    public static final String KAFKA_BROKER="master:9092,slave1:9092,slave2:9092";

    public static void main(String[] args) {
        new TridentKafkaTopology().run(args);
    }


    private void run(String[] args) {
        String brokerUrl = KAFKA_BROKER;

        System.out.printf("Running with broker_url: [%s], topics: [%s, %s]\n", brokerUrl, TOPIC1, TOPIC2);

        Config config = new Config();
//        config.setDebug(true);
        config.setMaxSpoutPending(4);
        config.setNumWorkers(4);

        //localsubmit
        final LocalSubmitter localSubmitter = LocalSubmitter.newInstance();
        final String topic1Tp = "topic1-producer";
        final String topic2Tp = "topic2-producer";
        final String consTpName = "topics-consumer";
        /**
         * 构建StormTopology时，使用TridentTopology来构建。newTopology方法
         * 这个localSubmitter.getDrpc()返回的是一个LocalDRPC,在本地调试的时候需要传，集群上不需要传。
         * 另一个参数是 ITridentDataSource ，代表一个数据源。
         * 这个KafkaTridentSpoutOpaque也是storm-kafka-client里提供的类。由KafkaSpoutConfig构建。
         * 在使用TopologyBuilder方式构建的原生拓扑中，也可以用这个直接KafkaSpoutConfig构建KafkaSpout
         */
        if(args.length>0 && "cluster".equals(args[0])){
            try {
                StormSubmitter.submitTopology(consTpName, config, TridentKafkaConsumerTopology.newTopology(newKafkaTridentSpoutOpaque()));
                Thread.sleep(2000);
                DrpcResultsPrinter.remoteClient().printResults(60, 1, TimeUnit.SECONDS);
            } catch (AlreadyAliveException e) {
                e.printStackTrace();
            } catch (InvalidTopologyException e) {
                e.printStackTrace();
            } catch (AuthorizationException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }else{
            //消费kafka的topic下所有历史消息，放入到TridentState中
            //这个示例中我去掉了往kafka发消息的拓扑，改用手动往kafka中发送消息。
            localSubmitter.submit(consTpName, config, TridentKafkaConsumerTopology.newTopology(
                    localSubmitter.getDrpc(), newKafkaTridentSpoutOpaque()));
            //调用DRPC服务，比对测试字符串的出现次数。
            localSubmitter.printResults(15, 1, TimeUnit.SECONDS);
        }

    }

    private KafkaTridentSpoutOpaque<String, String> newKafkaTridentSpoutOpaque() {
        KafkaSpoutConfig spoutConfig = newKafkaSpoutConfig();
        return new KafkaTridentSpoutOpaque<>(spoutConfig);
    }

    protected KafkaSpoutConfig<String,String> newKafkaSpoutConfig() {
        return KafkaSpoutConfig.builder(KAFKA_BROKER, TOPIC1)
                .setProp(ConsumerConfig.GROUP_ID_CONFIG, "kafkaSpoutTestGroup_" + System.nanoTime())
                .setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 200)
                //结果转换，将kafka消费到的value字段声明成value
                .setRecordTranslator(JUST_VALUE_FUNC, new Fields("topic","value"))
//                .setRecordTranslator(new DefaultRecordTranslator<>())
                //重试方法
                .setRetry(newRetryService())
                //提交offset间隔
                .setOffsetCommitPeriodMs(10_000)
                //启动时从该topic的最开始的消息进行消费。该示例中会由TridentKafkaConsumerTopology的addTridentState方法定义的stream消费
                //这个只跟启动时的消费策略有关。
                .setFirstPollOffsetStrategy(EARLIEST)
//                .setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy.LATEST)
                .setMaxUncommittedOffsets(250)
                .build();
    }

    private static Func<ConsumerRecord<String, String>, List<Object>> JUST_VALUE_FUNC = new JustValueFunc();

    private static class JustValueFunc implements Func<ConsumerRecord<String, String>, List<Object>>, Serializable {
        @Override
        public List<Object> apply(ConsumerRecord<String, String> record) {
            return new Values(record.topic(),record.value());
        }
    }

    protected KafkaSpoutRetryService newRetryService() {
        return new KafkaSpoutRetryExponentialBackoff(new TimeInterval(500L, TimeUnit.MICROSECONDS),
                TimeInterval.milliSeconds(2), Integer.MAX_VALUE, TimeInterval.seconds(10));
    }
}
